package gu.simplemq.jms;

import java.util.Iterator;
import java.util.concurrent.ConcurrentMap;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.Topic;


import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Maps;

import gu.simplemq.IAdvisor;

import static com.google.common.base.Preconditions.*;

/**
 * 通过订阅'ActiveMQ.Advisory.Consumer.Topic','ActiveMQ.Advisory.Consumer.Queue'统计每个频道/队列订阅/消费者数量(线程安全)
 * @author guyadong
 * @deprecated
 */
public class AdvisoryMessageManager implements AutoCloseable,IAdvisor,JmsConstants,MessageListener,JMSReconnectCallback{
	private final ConcurrentMap<String, Message> advisoryMessages = Maps.newConcurrentMap();
    public static final String QUEUE_CONSUMER_ADVISORY_TOPIC_PREFIX = "ActiveMQ.Advisory.Consumer.Queue.";
    public static final String TOPIC_CONSUMER_ADVISORY_TOPIC_PREFIX = "ActiveMQ.Advisory.Consumer.Topic.";

	private static final LoadingCache<JmsPoolLazy,AdvisoryMessageManager> CACHE = CacheBuilder.newBuilder().build(new CacheLoader<JmsPoolLazy, AdvisoryMessageManager>(){

		@Override
		public AdvisoryMessageManager load(JmsPoolLazy key) throws Exception {
			return new AdvisoryMessageManager(key);
		}});
	private volatile Connection advisoryConnection;
	private volatile Session advisorySession;
	private final JmsPoolLazy poolLazy;
	private final AutoReconnectAdapter autoReconnectAdapter = new AutoReconnectAdapter(this);
	private AdvisoryMessageManager(JmsPoolLazy poolLazy) {
		this.poolLazy = checkNotNull(poolLazy, "poolLazy is null");
		init();
	}
	private void init() {
		try {
			advisoryConnection = this.poolLazy.borrow();
			new ExceptionListenerContainer(autoReconnectAdapter).bind(advisoryConnection);
			advisoryConnection.start();
			advisorySession = advisoryConnection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
			sub(advisorySession,TOPIC_CONSUMER_ADVISORY_TOPIC_PREFIX + "*", this);
			sub(advisorySession,QUEUE_CONSUMER_ADVISORY_TOPIC_PREFIX + "*", this);
		} catch (JMSException e) {
			throw new RuntimeException(e);
		}
	}
	private void uninit() {
		synchronized (this) {
			if(null != advisoryConnection){
				this.poolLazy.release(advisoryConnection);
				advisoryConnection = null;
			}
		}
	}
	private static void sub(Session session,String topic,MessageListener listener) throws JMSException {
		Topic t = session.createTopic(topic);
		MessageConsumer consumer = session.createConsumer(t);
		consumer.setMessageListener(listener);
	}
    private static String getAdvisoryTopic(String destination, String prefix) {
        return prefix + destination.replaceAll(",", "&sbquo;");
    }

	/**
	 * 返回指定频道的 advisory message<br>
	 * @param name
	 * @return advisory message,没有收到消息返回{@code null}
	 */
    Message advisoryMessageOfTopic(String name){
		String advisoryTopic = getAdvisoryTopic(name, TOPIC_CONSUMER_ADVISORY_TOPIC_PREFIX);
		return advisoryMessages.get(advisoryTopic);
	}
	/**
	 * 返回指定队列的 advisory message<br>
	 * @param name
	 * @return advisory message,没有收到消息返回{@code null}
	 */
    Message advisoryMessageOfQueue(String name){
		String advisoryTopic = getAdvisoryTopic(name, QUEUE_CONSUMER_ADVISORY_TOPIC_PREFIX);
		return advisoryMessages.get(advisoryTopic);
	}
	@Override
	public void close() {
		uninit();
		CACHE.asMap().remove(poolLazy);
	}
	
	@Override
	public int consumerCountOf(String channelName) {
		try {
			Message message = advisoryMessageOfQueue(channelName);
			return message == null ? 0 : message.getIntProperty(PROP_CONSUMER_COUNT);
		} catch (JMSException e) {
			throw new RuntimeException(e);
		}
	}
	
	@Override
	public int subscriberCountOf(String channelName) {
		try {
			Message message = advisoryMessageOfTopic(channelName);
			return message == null ? 0 : message.getIntProperty(PROP_CONSUMER_COUNT);
		} catch (JMSException e) {
			throw new RuntimeException(e);
		}
	}
	@Override
	public void onMessage(Message message) {
		
		try {
			String dest = poolLazy.runtimeContext.destNameOf(message);
			advisoryMessages.put(dest, message);
		} catch (JMSException e) {
			e.printStackTrace();
		}
		
	}
	@Override
	public void onConnectionLost() throws Exception {
		uninit();
	}
	@Override
	public void tryReconnecting() throws Exception {
		init();	
	}
	@Override
	public String ownerName() {
		return getClass().getSimpleName();
	}
	public static AdvisoryMessageManager instanceOf(JmsPoolLazy pool){
		return CACHE.getUnchecked(pool);
	}
	/**
	 * 关闭并删除所有资源池中的{@link AdvisoryMessageManager}实例
	 */
	public synchronized static void closeAll(){
		for(Iterator<AdvisoryMessageManager> itor = CACHE.asMap().values().iterator();itor.hasNext();){
			AdvisoryMessageManager p = itor.next();
			itor.remove();
			p.close();
		}
	}

}
